//////////////////////////////////////////////////
// EASIM project
// easim.net.DtnNode
// 6/17/2007 5:48 PM
// Cong Liu
// gzcong@gmail.com

// Copyright (C) 2007
// All rights reserved.
// Refer to LICENSE of JIST (Java In Simulation Time) for terms and conditions of use.

package easim.net;

import jist.swans.misc.Message;
import jist.swans.Constants;
import jist.runtime.JistAPI.DoNotRewrite;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;

public abstract class DtnNode extends Node implements DtnNodeInterface, DoNotRewrite
{

  ////////////////////////////////
  // neighbor management
  //

  /** class ContactInfo */
  public class ContactInfo implements Timer
  {
    public int id;
    public void timeout() {
      contactTimeout(this);
    }
  } // end class ContactInfo

  protected HashMap contacts = new HashMap();

  protected void contactTimeout(ContactInfo contact) {
    //System.out.println(id + " contact timeout " + Clock.getTime());
    acksBox.remove(new Integer(contact.id));
    onContactBroken(contact.id);
    contacts.remove(new Integer(contact.id));
  }

  public void receive(Message msg, int srcId, int dstId) {
    //System.out.println(id + " receive " + srcId + "->" + dstId);
    ContactInfo contact = (ContactInfo)contacts.get(new Integer(srcId));
    if (contact == null) {
      contact = new ContactInfo();
      contact.id = srcId;
      contacts.put(new Integer(contact.id), contact);
      onContactDetected(contact.id);
    }
    PiggybackMessage pMsg = (PiggybackMessage)msg;
    processAcks(pMsg.acks, srcId, pMsg.msgId);
    if (pMsg.message != null) {
      receiveMessage(pMsg.message, srcId);
    }
    Clock.start(contact, CONTACT_TIMEOUT, true);
    //System.out.println(id + " reschedule contact timeout " + (Clock.getTime() + CONTACT_TIMEOUT));
  }

  //////////////////////////////
  // message scheduling
  // 

  public DtnNode(int id) {
    super(id);
  }

  public void start() {
    startJitterTimer();
  }

  public void pump(int queueSize, boolean macBusy) {
    if (queueSize == 0 && macBusy == false) {
      startJitterTimer();
    }
  }

  /// Jitter
  public void startJitterTimer() {
    long jitter = JITTER_PER_NEIGHBOR_MIN + (long)((JITTER_PER_NEIGHBOR_MAX - 
      JITTER_PER_NEIGHBOR_MIN) * Constants.random.nextDouble());
    jitter *= contacts.size();
    Clock.start(jitterTimer, jitter, false); // if not +1, 'sendNextMessage' might be called before <init> is done.
  }
  protected class JitterTimer implements Timer, DoNotRewrite
  {
    public void timeout() {
      sendNextMessage();
    }
  }
  protected JitterTimer jitterTimer = new JitterTimer();

  /// Hello Message
  protected void startHelloTimer() {
    isHelloTimerOn = true;
    long helloInterval = HELLO_INTERVAL_MIN + (long)((HELLO_INTERVAL_MAX - 
      HELLO_INTERVAL_MIN) * Constants.random.nextDouble());
    Clock.start(helloTimer, helloInterval, true);
  }
  protected class HelloTimer implements Timer, DoNotRewrite
  {
    public void timeout() {
      isHelloTimerOn = false;
      startJitterTimer();
    }
  }
  protected HelloTimer helloTimer = new HelloTimer();
  protected boolean isHelloTimerOn = false;

  protected void sendNextMessage() {
    Message msg = nextMessageToSend();
    if (msg == null) { // send hello message
      if (! isHelloTimerOn) {
        sendReliable(null);
        startHelloTimer();
      }
    }
    else { // send reliable broadcast
      sendReliable(msg);
      // restart hello timer after sending each message.
      startHelloTimer();
    }
  }

  //////////////////////////////
  // reliable broadcast
  //

  protected int msgId = 0;
  protected HashMap acksBox = new HashMap(); // Integer:HashSet<Integer>
  protected HashMap msgsBox = new HashMap(); // Integer:{PiggybackMessage,HashSet<Integer>}

  public class PiggybackMessage implements Message, Timer, DoNotRewrite
  {
    public int getSize() {
      int size = 4;
      if (message != null) {
        size += message.getSize();
      }
      Iterator it = acks.values().iterator();
      while (it.hasNext()) {
        HashSet msgIds = (HashSet)it.next();
        size += 4 * (msgIds.size() + 2);
      }
      return size;
    }
    public void getBytes(byte[] msg, int offset) {
      throw new RuntimeException("no implementation.");
    }
    public void timeout() {
      ackTimeout(this);
    }
    public int msgId;
    public Message message;
    public HashMap acks = new HashMap();
  }

  protected void processAcks(HashMap acks, int srcId, int msgId) {
    // add an ack to the acksBox, which will be send in the next opportunity
    HashSet acksBox1 = (HashSet)acksBox.get(new Integer(srcId));
    if (acksBox1 == null) {
      acksBox1 = new HashSet();
      acksBox.put(new Integer(srcId), acksBox1);
    }
    acksBox1.add(new Integer(msgId));
    // handle asks and clear up acksBox.
    HashSet acks1 = (HashSet)acks.get(new Integer(this.id));
    if (acks1 == null) return;
    Iterator it = acks1.iterator();
    while (it.hasNext()) {
      msgId = ((Integer)it.next()).intValue();
      Object[] objs = (Object[])msgsBox.get(new Integer(msgId));
      if (objs == null) {
        continue;
      }
      // ack local msgs.
      PiggybackMessage pMsg = (PiggybackMessage)objs[0];
      HashSet unAcked = (HashSet)objs[1];
      unAcked.remove(new Integer(srcId));
      if (unAcked.size() == 0) {
        Clock.cancel(pMsg);
        msgsBox.remove(new Integer(pMsg.msgId));
        if (pMsg.message != null) {
          onAckMessage(pMsg.message);
        }
      }
      // clean up acksBoxs
      HashSet acks2 = (HashSet)pMsg.acks.get(new Integer(srcId));
      if (acks2 != null) {
        acksBox1.removeAll(acks2);
        if (acksBox1.size() == 0) {
          acksBox.remove(new Integer(srcId));
        }
      }
    }
  }

  protected HashMap generateAcks() {
    HashMap asks = new HashMap();
    Iterator it = acksBox.keySet().iterator();
    while (it.hasNext()) {
      Integer dstId = (Integer)it.next();
      HashSet acks1 = (HashSet)acksBox.get(dstId);
      asks.put(dstId, new HashSet(acks1));
    }
    return asks;
  }

  protected void ackTimeout(PiggybackMessage pMsg) {
    Object[] objs = (Object[])msgsBox.remove(new Integer(pMsg.msgId));
    HashSet unAcked = (HashSet)objs[1];
    if (unAcked.size() == 0) throw new RuntimeException("impossible situation.");
    if (pMsg.message == null) return;
    int[] dstIds = new int[unAcked.size()];
    Iterator it = unAcked.iterator();
    for (int i = 0; i < dstIds.length; ++ i) {
      Integer dstId = (Integer)it.next();
      dstIds[i] = dstId.intValue();
    }
    if (pMsg.message != null) {
      onDropMessage(pMsg.message, dstIds);
    }
  }

  protected void sendReliable(Message msg) {
    PiggybackMessage pMsg = new PiggybackMessage();
    pMsg.msgId = ++ msgId;
    pMsg.message = msg;
    HashSet unAsked = new HashSet(contacts.keySet());
    if (unAsked.size() > 0) {
      pMsg.acks = generateAcks();
      Clock.start(pMsg, ACK_TIMEOUT, false);
      msgsBox.put(new Integer(pMsg.msgId), new Object[] {pMsg, unAsked});
    }
    //int size = msg.getSize();
    send(pMsg, ADDRESS_ANY);
    //System.out.println(id + " send " + Clock.getTime());
    //return size;
  }

}
